package cn.ac.iie.bdb_to_mongo.sender;

import cn.ac.iie.bdb_to_mongo.BdbMain;
import cn.ac.iie.bdb_to_mongo.commons.Constants;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.REMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.LinkedBlockingDeque;

/**
 * @auther Fighter Created on 2016/12/10.
 */
public class RewriteSenderThread extends Thread {

    private static final Logger LOG = LoggerFactory.getLogger(RewriteSenderThread.class);
    MessageSender sender;
    long timeOut;
    long batchNum;
    List<Object[]> cacheDatas = new ArrayList<>();
    LinkedBlockingDeque<REMessage> queue;

    public RewriteSenderThread(String nameServer, String topic, LinkedBlockingDeque<REMessage> queue) {
        sender = new MessageSender(nameServer, topic);
        timeOut = BdbMain.configuration.getLong(Constants.SENDER_QUEUE_TIME_OUT, 5000);
        batchNum = BdbMain.configuration.getLong(Constants.SENDER_QUEUE_BATCH_NUM, 10);
        this.queue = queue;
    }

    @Override
    public void run() {
        LOG.info(" rewrite send thread start!");
        long startTime = System.currentTimeMillis();
        while (true) {
            try {
                REMessage message = this.queue.take();
                if (message == null) {
                    LOG.warn(" message is null!");
                    continue;
                }
                List<Object[]> datas = message.data;
                if (datas == null || datas.isEmpty()) {
                    LOG.warn(" message datas is null!");
                    continue;
                }
                cacheDatas.addAll(datas);
                long endTime = System.currentTimeMillis();
                long time = endTime - startTime;
                if (time < timeOut && cacheDatas.size() < batchNum) {
                    continue;
                }
                REMessage cacheMessage = new REMessage(
                        message.colNames,
                        message.types,
                        cacheDatas,
                        message.docSchemaName,
                        message.userDesc);
                sender.sendMessage(cacheMessage);
                LOG.info("rewrite send messages num: {}", cacheDatas.size());
                cacheDatas.clear();
                startTime = System.currentTimeMillis();
            } catch (Exception ex) {
                LOG.error("rewrite send messages error!", ex);
            }
        }
    }
}
